﻿Imports MySql.Data
Friend Class dbiMySql
    Implements iDBInfo

    Private _connection As MySqlClient.MySqlConnection
    Private _connectionName As String
    Private _messages As New Queue(Of PSObject)
    Private _transaction As MySqlClient.MySqlTransaction

    Public Property CommandTimeout As Integer Implements iDBInfo.CommandTimeout

    Public ReadOnly Property Connection As IDbConnection Implements iDBInfo.Connection
        Get
            Return Me._connection
        End Get
    End Property

    Public ReadOnly Property ConnectionInfo As PSObject Implements iDBInfo.ConnectionInfo
        Get
            Dim PS As New PSObject
            With PS.Members
                .Add(New PSNoteProperty("ConnectionName", Me.ConnectionName))
                .Add(New PSNoteProperty("ConnectionType", Me.Type.ToString))
                .Add(New PSNoteProperty("ConnectionState", Me.Connection.State))
                .Add(New PSNoteProperty("ConnectionString", Me.Connection.ConnectionString))
                .Add(New PSNoteProperty("ConnectionTimeout", Me.Connection.ConnectionTimeout))
                .Add(New PSNoteProperty("ServerVersion", Me._connection.ServerVersion))
                .Add(New PSNoteProperty("Database", Me._connection.Database))
                .Add(New PSNoteProperty("CommandTimeout", Me.CommandTimeout))
                .Add(New PSNoteProperty("HasTransaction", Me._transaction IsNot Nothing))
            End With
            Return PS
        End Get
    End Property

    Public ReadOnly Property ConnectionName As String Implements iDBInfo.ConnectionName
        Get
            Return Me._connectionName
        End Get
    End Property

    Friend Function CreateMap(dt As DataTable) As List(Of DBInfo.Map) Implements iDBInfo.CreateMap
        Return dt.AsEnumerable.Select(Function(x) New DBInfo.Map(x("ColumnOrdinal"), x("Columnname"), x("DataType"), x("AllowDBNull"))).ToList
    End Function

    Public Function GetCommand(query As String, commandTimeout As Integer, Optional parameters As Hashtable = Nothing) As IDbCommand Implements iDBInfo.GetCommand
        If commandTimeout < 0 Then commandTimeout = Me.CommandTimeout
        Dim cmd As New MySqlClient.MySqlCommand(query, Me._connection)
        cmd.CommandTimeout = commandTimeout
        If parameters IsNot Nothing Then
            For Each kvp As DictionaryEntry In parameters
                If kvp.Value Is Nothing Then
                    cmd.Parameters.AddWithValue(kvp.Key, DBNull.Value)
                Else
                    cmd.Parameters.AddWithValue(kvp.Key, kvp.Value)
                End If
            Next
        End If
        Return cmd
    End Function

    Public Function GetCommand(query As String, Optional parameters As Hashtable = Nothing) As IDbCommand Implements iDBInfo.GetCommand
        Return GetCommand(query, Me.CommandTimeout, parameters)
    End Function

    Public Function GetDataSet(cmd As IDbCommand) As DataSet Implements iDBInfo.GetDataSet
        Dim ds As New DataSet
        Using da As New MySqlClient.MySqlDataAdapter(cmd)
            da.Fill(ds)
        End Using
        Return ds
    End Function

    Public Function GetParameter() As IDataParameter Implements iDBInfo.GetParameter
        Return New MySqlClient.MySqlParameter
    End Function

    Friend ReadOnly Property Type As DBInfo.DBInfoType Implements iDBInfo.Type
        Get
            Return DBInfo.DBInfoType.MySql
        End Get
    End Property

    Public Sub New(name As String, server As String, port As Integer, database As String, userid As String, password As String, timeout As Integer)
        Me._connectionName = name
        Me.CommandTimeout = timeout

        Dim cb As New MySqlClient.MySqlConnectionStringBuilder
        With cb
            .Server = server
            .Database = database
            .UserID = userid
            .Password = password
            .Port = port
        End With
        Me._connection = New MySqlClient.MySqlConnection(cb.ToString)

        Call Me.Init()
    End Sub

    Public Sub New(name As String, connStr As String, timeout As Integer)
        Me._connectionName = name
        Me.CommandTimeout = timeout
        Me._connection = New MySqlClient.MySqlConnection(connStr)

        Call Me.Init()
    End Sub

    Private Sub Init()
        Try
            Me._connection.Open()
        Catch ex As Exception
            Me._connection.Dispose()
            Me._connection = Nothing
            Throw
        End Try
    End Sub

    Public Sub ClearMessages() Implements iDBInfo.ClearMessages
    End Sub

    Public Function GetMessage() As PSObject Implements iDBInfo.GetMessage
        Throw New NotImplementedException("SQLIte does not support messages.")
    End Function

    Public ReadOnly Property HasMessages As Boolean Implements iDBInfo.HasMessages
        Get
            Throw New NotImplementedException("SQLIte does not support messages.")
        End Get
    End Property

    Public Function BulkLoad(dataReader As IDataReader, destinationTable As String, columnMap As Hashtable, batchSize As Integer, batchTimeout As Integer, notify As Action(Of Integer)) As Long Implements iDBInfo.BulkLoad
        Dim typeMap As List(Of DBInfo.Map) = Me.CreateMap(dataReader.GetSchemaTable)
        If columnMap IsNot Nothing Then
            typeMap.RemoveAll(Function(x) Not columnMap.ContainsKey(x.Name))
            typeMap.ForEach(Sub(x) x.MappedName = columnMap(x.Name))
        Else
            typeMap.ForEach(Sub(x) x.MappedName = x.Name)
        End If

        Dim insertSql As String = String.Format("INSERT INTO {0} ({1}) VALUES ({2})", destinationTable, String.Join(", ", typeMap.Select(Function(x) x.MappedName)), "@" & String.Join(", @", typeMap.Select(Function(x) x.MappedName)))

        Using cmd As New MySqlClient.MySqlCommand(insertSql, Me._connection)
            cmd.Transaction = Me._connection.BeginTransaction()
            Dim sw As New Stopwatch
            Dim batchIteration As Integer = 0
            sw.Start()
            Try
                While dataReader.Read()
                    batchIteration += 1
                    cmd.Parameters.Clear()
                    typeMap.ForEach(Sub(x) cmd.Parameters.AddWithValue(x.MappedName, dataReader.GetValue(x.Ordinal)))
                    cmd.ExecuteNonQuery()
                    If sw.Elapsed.TotalSeconds > batchTimeout Then
                        Throw New TimeoutException(String.Format("Batch took longer than {0} seconds to complete.", batchTimeout))
                    End If
                    If batchIteration Mod batchSize = 0 Then
                        cmd.Transaction.Commit()
                        cmd.Transaction.Dispose()
                        If notify IsNot Nothing Then notify(batchIteration)
                        cmd.Transaction = Me._connection.BeginTransaction
                        sw.Restart()
                    End If
                End While
                cmd.Transaction.Commit()
                cmd.Transaction.Dispose()
                sw.Stop()
            Catch ex As Exception
                If cmd.Transaction IsNot Nothing Then
                    cmd.Transaction.Rollback()
                    cmd.Transaction.Dispose()
                End If
                Throw ex
            End Try
            BulkLoad = batchIteration
        End Using
    End Function

    Public Sub BeginTransaction() Implements iDBInfo.BeginTransaction
        If Me._transaction IsNot Nothing Then Throw New InvalidOperationException("Cannot BEGIN a transaction when one is already begun.")
        Me._transaction = Me._connection.BeginTransaction
    End Sub

    Public Sub CommitTransaction() Implements iDBInfo.CommitTransaction
        If Me._transaction Is Nothing Then Throw New InvalidOperationException("Cannot COMMIT a transaction when one does not exist.")
        Me._transaction.Commit()
        Me._transaction.Dispose()
        Me._transaction = Nothing
    End Sub

    Public Sub RollbackTransaction(Optional silent As Boolean = False) Implements iDBInfo.RollbackTransaction
        If Me._transaction IsNot Nothing Then
            Me._transaction.Rollback()
            Me._transaction.Dispose()
            Me._transaction = Nothing
        Else
            If Not silent Then Throw New InvalidOperationException("Cannot ROLLBACK a transaction when one does not exist.")
        End If
    End Sub

    Public Sub AttachTransaction(command As IDbCommand) Implements iDBInfo.AttachTransaction
        command.Connection = Me._connection
        If Me._transaction IsNot Nothing Then command.Transaction = Me._transaction
    End Sub
End Class
